package work.base.data.queue;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

// 1. 定义事件类
class LongEvent {
    private long value;

    public void set(long value) {
        this.value = value;
    }

    @Override
    public String toString() {
        return "LongEvent{value=" + value + "}";
    }
}

// 2. 定义事件工厂
class LongEventFactory implements EventFactory<LongEvent> {
    @Override
    public LongEvent newInstance() {
        return new LongEvent();
    }
}

// 3. 定义事件处理器（消费者）
class LongEventHandler implements EventHandler<LongEvent> {
    @Override
    public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
        System.out.println("消费者处理事件: " + event);
    }
}

// 4. 定义生产者
class LongEventProducer {
    private final RingBuffer<LongEvent> ringBuffer;

    public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void onData(long data) {
        // 获取下一个可用位置的序列号
        long sequence = ringBuffer.next();
        try {
            // 获取该位置对应的事件对象
            LongEvent event = ringBuffer.get(sequence);
            // 设置事件数据
            event.set(data);
        } finally {
            // 发布事件
            ringBuffer.publish(sequence);
        }
    }
}

// 5. 主程序
public class DisruptorExample {
    public static void main(String[] args) throws InterruptedException {
        // 创建线程池
        ExecutorService executor = Executors.newSingleThreadExecutor();

        // 创建事件工厂
        LongEventFactory factory = new LongEventFactory();

        // 设置环形缓冲区大小，必须是2的幂
        int bufferSize = 1024;

        // 创建Disruptor实例
        Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, executor);

        // 连接消费者
        disruptor.handleEventsWith(new LongEventHandler());

        // 启动Disruptor
        RingBuffer<LongEvent> ringBuffer = disruptor.start();

        // 创建生产者
        LongEventProducer producer = new LongEventProducer(ringBuffer);

        // 发布事件
        for (long i = 0; i < 10; i++) {
            producer.onData(i);
            Thread.sleep(100);
        }

        // 关闭Disruptor和线程池
        disruptor.shutdown();
        executor.shutdown();
    }
}

